package org.melanesia.sql;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.logging.Logger;

import org.melanesia.beans.BeanUtils;
import org.melanesia.conf.Config;
import org.melanesia.converters.JavaToJavaConverter;
import org.melanesia.converters.JavaToJdbcConverter;
import org.melanesia.converters.TypeConverter;
import org.melanesia.ref.ReferenceMonitor;
import org.melanesia.sql.PaginationSupport.SQLVariation;
import org.melanesia.sql.exceptions.DataBindingException;
import org.melanesia.sql.exceptions.SQLExecutionException;
import org.melanesia.sql.utils.JDBCUtils;
import org.melanesia.sql.utils.NamedParametersHelper;

/**
 * Query class allows for sql-query execution.
 *
 * @author marcin.kielar
 */
public final class Query extends AbstractSQL {

    /** Logger. */
    private static Logger logger = Logger.getLogger(Query.class.getName());

    /** Query medadata. */
    private final QueryMetaData queryMetaData = new QueryMetaData();

    /**
     * Intermediate property classes, for instantiating intermediate properties.
     */
    private final Map<String, Class<?>> itermediatePropertyClasses = new HashMap<String, Class<?>>();

    /**
     * Bean class factory.
     *
     * @see #setBeanClassFactory(BeanClassFactory)
     * @see Iterator#next()
     */
    private BeanClassFactory beanClassFactory;

    /**
     * Range callback.
     *
     * @see #setRangeCallback(RangeCallback)
     */
    private RangeCallback rangeCallback = null;

    /**
     * Helper class which determines the way pagination is handled by
     * {@code Query} and {@link Iterator} objects.
     *
     * @see #setPaginationMode(PaginationMode)
     */
    private PaginationMode paginationMode = PaginationMode.ITERATION_BASED;

    /**
     * Helper class which handles pagination related activities like wrapping
     * SQL for pagination, defining additional input parameters, etc.
     *
     * @see #setPaginationSupportClass(Class)
     */
    private Class<? extends PaginationSupport> paginationSupportClass;

    /**
     * Constructs new <code>Query</code>.
     *
     * @param connection
     *            jdbc connection
     * @param typeConverter
     *            type converter
     * @param sql
     *            sql clause
     */
    Query(final Connection connection, final TypeConverter typeConverter, final String sql) {
        super(connection, typeConverter, sql);
    }

    /**
     * Sets the query parameter value.
     *
     * @param name
     *            parameter name
     * @param value
     *            parameter value
     * @return this query
     */
    public Query setParameter(final String name, final Object value) {
        queryMetaData.setInputParameter(name, value);
        return this;
    }

    /**
     * Sets the query parameter value and explicitly defines java do jdbc
     * converter.
     *
     * @param name
     *            parameter name
     * @param value
     *            parameter value
     * @param converter
     *            java to jdbc converter
     * @return this query
     */
    public Query setParameter(final String name, final Object value, final JavaToJdbcConverter converter) {
        queryMetaData.setInputParameter(name, value, converter);
        return this;
    }

    /**
     * Registers a converter to be used to convert a value of specified column
     * from the result set.
     *
     * @param columnName
     *            name of the column alias to associate the converter with
     * @param converter
     *            converter to associate with the column alias
     * @return this query
     */
    public Query setColumnConverter(final String columnName, final JavaToJavaConverter converter) {
        queryMetaData.setColumnConverter(columnName, converter);
        return this;
    }

    /**
     * Tells melanesia to ignore given column from the result set, during mapping to result bean.
     * For such column, no property accessor is looked up, and no actions are taken to map the column to the bean's property.
     *
     * @param columnName column name to ignore
     * @return this query
     */
    public Query ignoreColumn(final String columnName) {
        queryMetaData.setIgnoredResult(columnName);
        return this;
    }


    /**
     * @param columnName column name to ignore
     * @return this query
     *
     * @deprecated use {@link #ignoreColumn(String)} instead.
     */
    @Deprecated
    public Query setIgnoredColumn(final String columnName) {
        ignoreColumn(columnName);
        return this;
    }

    /**
     * See {@link QueryMetaData#setPropertyPathForResult(String, String)}.
     *
     * @param columnName
     *            column name
     * @param propertyPath
     *            property path in result java bean
     * @return this query
     */
    public Query setPropertyPathForColumn(final String columnName, final String propertyPath) {
        queryMetaData.setPropertyPathForResult(columnName, propertyPath);
        return this;
    }

    /**
     * Sets class for a particular bean property. This is mainly used with
     * composite properties. The idea of composite properties is discussed in
     * {@link org.melanesia.beans.Setter#set(Object, Object, Map)}.
     *
     * @param bean
     *            name of the beanProperty used in SQL statement. for example:
     *            "b", "b.c", "b.c.d"
     * @param beanClass
     *            class to be used when instantiating the bean
     * @return this query
     */
    public Query setIntermediatePropertyClass(final String bean, final Class<?> beanClass) {
        itermediatePropertyClasses.put(bean, beanClass);
        return this;
    }

    /**
     * Registers a range callback. Range callback is an object holding
     * information about:
     *
     * <ul>
     * <li>start of the range to return</li>
     * <li>stop of the range to return</li>
     * <li>total count of rows in result set</li>
     * </ul>
     *
     * The value of total count of rows in result set is passed into the
     * {@code RangeCallback} using {@link RangeCallback#setTotalRows(long)}.
     *
     * @param rangeCallback range callback
     * @return this query
     */
    public Query setRangeCallback(final RangeCallback rangeCallback) {
        this.rangeCallback = rangeCallback;
        return this;
    }

    /**
     * Sets pagination mode of this query.
     *
     * @see PaginationMode for list of supported pagination mode.
     * @param paginationMode
     *            pagination mode
     * @return this query
     */
    public Query setPaginationMode(final PaginationMode paginationMode) {
        this.paginationMode = paginationMode;
        return this;
    }

    /**
     * Sets pagination support class for query-based pagination.
     *
     * @see PaginationMode#QUERY_BASED
     * @param paginationSupportClass
     *            pagination support class
     * @return this query
     */
    public Query setPaginationSupportClass(final Class<? extends PaginationSupport> paginationSupportClass) {
        this.paginationSupportClass = paginationSupportClass;
        return this;
    }

    /**
     * Registers a bean class factory with this query.
     * The factory is used during execution of the query in order to obtain specific subclasses
     * of the returnBeanClass depending on the result set data.
     *
     * @param beanFactory bean factory
     * @return this query
     */
    public Query setBeanClassFactory(final BeanClassFactory beanFactory) {
    	this.beanClassFactory = beanFactory;
    	return this;
    }

    /**
     * Executes the query with no result returned.
     */
    public void execute() {
        executeInternal(null, 0, null);
    }

    /**
     * Executes the query returning list of beans of class
     * <code>returnedBeanClass</code>. Each row from the result set is mapped
     * into single instance of <code>returnedBeanClass</code>.
     *
     * @param <T>
     *            returned type
     * @param returnedBeanClass
     *            returned class
     * @return list of beans representing the query result set
     */
    public <T> List<T> execute(final Class<T> returnedBeanClass) {
        return executeInternal(returnedBeanClass, 0, null);
    }

    /**
     * Executes the query returning the list of only the values of the selected
     * column.
     *
     * NOTE! This method SHOULD be given an Object type (not a primitive type)
     * in the <code>returnedClass</code>.
     *
     * If a primitive type is given (i.e. int.class, boolean.class), and there
     * is no result, or the result is null, this method will cause an
     * {@link org.melanesia.beans.exceptions.NullCastToPrimitiveTypeException},
     * as described in {@link org.melanesia.beans.BeanUtils#cast(Object, Class)}
     *
     * The user MAY pass a primitive type if she/he is sure that there will
     * always be a not null result or if she/he is aware of the
     * {@link org.melanesia.beans.exceptions.NullCastToPrimitiveTypeException}
     * and handles it in user code.
     *
     * It is however advised not to use primitive types at all with this method.
     *
     * @param <T>
     *            return type
     * @param returnedClass
     *            type of returned values
     * @param returnColumnName
     *            column name from result set, for which value will be returned
     * @return list of values for given <code>returnColumnName</code>
     */
    public <T> List<T> execute(final Class<T> returnedClass, final String returnColumnName) {
        return executeInternal(returnedClass, 0, returnColumnName);
    }

    /**
     * Executes the query returning a bean of class
     * <code>returnedBeanClass</code> representing the first row of the result
     * set. This method should be used primarily in cases, when it is
     * guaranteed, that the query returns only one row, or it is desired to
     * process only the first row of a result set.
     *
     * NOTE! For optimization, it is advised to use "AND ROWNUM = 1", or similar
     * mechanism guaranteeing the return of a single row.
     *
     * @param <T>
     *            returned type
     * @param returnedBeanClass
     *            returned class
     * @return list of beans representing the query result set
     */
    public <T> T executeSingle(final Class<T> returnedBeanClass) {

        List<T> list = executeInternal(returnedBeanClass, 1, null);
        if (!list.isEmpty()) {
            return list.get(0);
        } else {
            return null;
        }
    }

    /**
     * Executes the query returning only the value of the selected column from
     * first row of the result set. This method should be used primarily in
     * cases, when it is guaranteed, that the query returns only one row, or it
     * is desired to process only the first row of a result set.
     *
     * NOTE! For optimization, it is advised to use "AND ROWNUM = 1", or similar
     * mechanism guaranteeing the return of a single row.
     *
     * NOTE! This method SHOULD be given an Object type (not a primitive type)
     * in the <code>returnedClass</code>.
     *
     * If a primitive type is given (i.e. int.class, boolean.class), and there
     * is no result, or the result is null, this method will cause an
     * {@link org.melanesia.beans.exceptions.NullCastToPrimitiveTypeException},
     * as described in {@link org.melanesia.beans.BeanUtils#cast(Object, Class)}
     *
     * The user MAY pass a primitive type if she/he is sure that there will
     * always be a not null result or if she/he is aware of the
     * {@link org.melanesia.beans.exceptions.NullCastToPrimitiveTypeException}
     * and handles it in user code.
     *
     * It is however advised not to use primitive types at all with this method.
     *
     * @param <T>
     *            object type of returned value
     * @param returnedClass
     *            returned class
     * @param returnColumnName
     *            name of the column from result set to return
     * @return value of the column specified in <code>returnColumnName</code>
     */
    public <T> T executeSingle(final Class<T> returnedClass, final String returnColumnName) {

        List<T> list = executeInternal(returnedClass, 1, returnColumnName);
        if (!list.isEmpty()) {
            return list.get(0);
        } else {
            return null;
        }
    }

    /**
     * Works similar to {@link #execute(Class)}, except it returns an
     * {@link Iterator} instead of the list of results. The difference is this
     * method does not fetch all the rows of the result set immediately. The
     * rows are fetched (and mapped to resulting beans) only when
     * {@link Iterator#next} is called.
     *
     * This approach was designed to minimize memory usage - when the rows of
     * the result set are iterated, only a single bean exists at specific moment
     * of time.
     *
     * @param <T>
     *            return type
     * @param returnedClass
     *            returned class
     * @return result set iterator
     */
    public <T> Iterator<T> iterator(final Class<T> returnedClass) {
        return iterator(returnedClass, null);
    }

    /**
     * This method is similar to {@link #iterator(Class)} except that it does
     * not map the whole result set to a result bean, but returns a list of only
     * the single column values. The column name is specified by
     * {@code returnColumnName}.
     *
     * @param <T>
     *            return type
     * @param returnedClass
     *            returned class
     * @param returnColumnName
     *            column name from result set, for which value will be returned
     * @return result set iterator accessing single column
     */
    public <T> Iterator<T> iterator(final Class<T> returnedClass, final String returnColumnName) {
        try {
            return new Iterator<T>(returnedClass, returnColumnName, 0);
        } catch (Throwable t) {
            throw new SQLExecutionException(t);
        }
    }

    /**
     * Internal method for executing queries.
     *
     * @param <T>
     *            return type
     * @param returnedClass
     *            return type
     * @param maxFetchSize
     *            maximum fetch size - callers should set this value depending
     *            on expected / needed size of the result set
     * @param returnColumnName
     *            name of the column, for single column retrieval
     * @return list of query results
     */
    private <T> List<T> executeInternal(final Class<T> returnedClass, final int maxFetchSize, final String returnColumnName) {

        ArrayList<T> result = new ArrayList<T>();
        Iterator<T> iterator = null;
        try {

            iterator = new Iterator<T>(returnedClass, returnColumnName, maxFetchSize);

            if (returnedClass != null) {

                int cnt = 0;

                while (iterator.hasNext() && (maxFetchSize <= 0 || maxFetchSize > cnt)) {
                    T val = iterator.next();
                    result.add(val);

                    ++cnt;
                }

                if (cnt == 0 && returnColumnName != null) {
                    // We have a specified output parameter to return,
                    // but query returned no result. We have to return something
                    // anyway, so try null binding to be sure it works.
                    T acceptedValue = bindResultToSimpleValue(returnedClass, returnColumnName, null, null);

                    result.add(acceptedValue);
                }

            }

            return result;

        } catch (SQLExecutionException ex) {
            throw ex;
        } catch (Throwable t) {
            throw new SQLExecutionException(t);
        } finally {
            closeIterator(iterator);
        }
    }

    /**
     * Iterator wrapper for a result set. This class implements the
     * {@link Iterator} interface, but extends it with a
     * {@link Iterator#close()} method. The {@link Iterator#close()} should be
     * called always when the {@code Iterator} will no longer be used, as it
     * closes internal {@link ResultSet} and {@link java.sql.Statement}.
     *
     * Note, that the {@code Iterator} is automatically closed when all of the
     * internal result set rows are returned using {@link #next} method (so the
     * {@link #hasNext()} method returns false). This happens when the
     * {@code Iterator} is used internally, like in the
     * {@link Query#execute(Class)} method. However if one uses
     * {@link Query#iterator(Class)} methods the iterator is exported from the
     * {@link Query} object and must be closed explicitly when done using.
     *
     * @author marcin.kielar
     *
     * @param <T>
     *            value class for this iterator
     */
    public final class Iterator<T> implements java.util.Iterator<T> {

        /** The statement. */
        private final PreparedStatement statement;

        /** The result set. */
        private final ResultSet resultSet;

        /** The result set metadata. */
        private final ResultSetMetaData resultSetMetadata;

        /** The bean class to map results to. */
        private final Class<T> returnedClass;

        /** A column name from result set, which value is to be returned. */
        private final String returnColumnName;

        /** Current row of the result set. */
        private long currentRow = 0;

        /** Starting row for a range query. */
        private long startRow = 1;

        /** Stopping row for a range query. @{code -1} means "go to the end of the result set". */
        private long stopRow = -1;

        /** A flag indicating there are more records in the iterator. */
        private Boolean hasNext = null;

        /** A flag indicating if this iterator has already been closed. */
        private boolean isClosed = false;

        /** Pagination support for query-based ranges. */
        private PaginationSupport paginationSupport = null;

        /**
         * Constructs new {@code Iterator}.
         *
         * @param returnedClass
         *            the bean class to map results to
         * @param returnColumnName
         *            a column name from result set, which value is to be
         *            returned
         * @param maxFetchSize
         *            maximum fetch size - callers should set this value
         *            depending on expected / needed size of the result set
         */
        Iterator(final Class<T> returnedClass, final String returnColumnName, final int maxFetchSize) {

            try {

                this.returnedClass = returnedClass;
                this.returnColumnName = returnColumnName;

                this.startRow = rangeCallback == null ? 1 : rangeCallback.getStartRow();
                this.stopRow = rangeCallback == null ? -1 : rangeCallback.getStopRow();

                NamedParametersHelper named = null;
                List<InputParameter> additionalParameters = new ArrayList<InputParameter>();

                if (isPaginated() && paginationMode == PaginationMode.QUERY_BASED) {

                    paginationSupport = createPaginationSupport();

                    additionalParameters.addAll(paginationSupport.getAdditionalParameters(PaginationSupport.SQLVariation.RANGED_QUERY));
                    named = new NamedParametersHelper(paginationSupport.getSql(PaginationSupport.SQLVariation.RANGED_QUERY));

                } else {

                    named = getNamedParametersHelper();

                }

                if (Config.isDebugMode()) {
                    logger.info("Executing query:\n" + named.getSql());
                }

                statement = getConnection().prepareCall(named.getSql());

                bindInputParameters(statement, named);

                // Binding additional parameters (for query-based range pagination)
                if (paginationMode == PaginationMode.QUERY_BASED && !additionalParameters.isEmpty()) {
                    for (InputParameter parameter : additionalParameters) {
                        bindInputParameter(statement, parameter, named);
                    }
                }

                if (maxFetchSize > 0) {
                    statement.setFetchSize(maxFetchSize);
                } else {
                    statement.setFetchSize(Config.getQueryFetchSize());
                }

                long tstart = System.currentTimeMillis();
                resultSet = statement.executeQuery();


                if (Config.isDebugMode()) {
                    logger.info("Executing query took: " + (System.currentTimeMillis() - tstart) + "ms.");
                }

                resultSetMetadata = resultSet.getMetaData();

                if (Config.isDebugMode()) {

                    // Observe this iterator for cases when users do not call close
                    // at the end.
                    // Do this only in debug mode due to big influence on
                    // efficiency, especially on JBoss.

                    ReferenceMonitor.observe(this,
                            JDBCUtils.createFinalizer(resultSet),
                            JDBCUtils.createFinalizer(statement));
                }

            } catch (Throwable t) {
                throw new SQLExecutionException(t);
            }
        }

        /**
         * Creates new pagination support class.
         *
         * @see Config#getPaginationSupportClass()
         *
         * @return pagination support class for this query
         *
         * @throws Exception
         *             when instantiation of the class fails
         */
        private PaginationSupport createPaginationSupport() throws Exception {

            Class<? extends PaginationSupport> clazz = paginationSupportClass != null
                    ? paginationSupportClass
                    : Config.getPaginationSupportClass();

            PaginationSupport support = clazz.newInstance().initialize(getOriginalSql(), rangeCallback);

            return support;
        }

        /**
         * Closes this result iterator.
         */
        public void close() {

            try {

                if (!isClosed) {

                    if (rangeCallback != null) {

                        long totalRows = 0;

                        if (paginationMode == PaginationMode.ITERATION_BASED) {

                            // Iterate to the end of the result set (if not already iterated)
                            while (hasNext) {
                                hasNext = resultSet.next();
                                if (hasNext) {
                                    ++currentRow;
                                }
                            }

                            totalRows = currentRow;

                            JDBCUtils.close(resultSet);
                            JDBCUtils.close(statement);

                        } else if (paginationMode == PaginationMode.QUERY_BASED) {

                            // We close the statement and result set first, to allow execution of SELECT COUNT() later
                            JDBCUtils.close(resultSet);
                            JDBCUtils.close(statement);

                            // Optimization: if range was [1, infinity], and we iterated to the very end, we do not need to call COUNT.
                            if (!isPaginated() && !hasNext) {
                                totalRows = currentRow;
                            } else {
                                // Execute select count(1) from (...) on original query
                                totalRows = getTotalRowsWithSelectCount();
                            }

                        }

                        rangeCallback.setTotalRows(totalRows);

                    } else {

                        // No pagination, just close JDBC objects
                        JDBCUtils.close(resultSet);
                        JDBCUtils.close(statement);

                    }

                    isClosed = true;
                }

            } catch (Throwable t) {
                throw new SQLExecutionException(t);
            }

        }

        /**
         * Moves the internal result set forward by one row. If the iterator is
         * "ranged", first call moves resultSet to startRow, and when iterated
         * to stopRow, iterator is closed.
         */
        private void doNext() {

            try {

                if (hasNext == null && paginationMode == PaginationMode.ITERATION_BASED) {
                    // First call - the result set is positioned before first row, and we have iteration based pagination.
                    // That means, that we have to iterate to first requested row (in RangeCallback).
                    while (currentRow < startRow) {
                        hasNext = resultSet.next();
                        if (hasNext) {
                            ++currentRow;
                        } else {
                            break;
                        }
                    }
                } else {
                    hasNext = resultSet.next();
                    if (hasNext) {
                        ++currentRow;
                    }
                }

                if (stopRow > 0 && currentRow > stopRow && paginationMode == PaginationMode.ITERATION_BASED) {
                    // Iterated to stopRow - iterate to the end, and (by eventually setting hasNext to false)
                    // force closing of the result set in the condition below.
                    while (hasNext) {
                        hasNext = resultSet.next();
                        if (hasNext) {
                            ++currentRow;
                        }
                    }
                }

                if (!hasNext) {
                    close();
                }

            } catch (Throwable t) {
                throw new SQLExecutionException(t);
            }
        }

        @Override
        public boolean hasNext() {

            if (hasNext == null) {
                doNext();
            }

            return hasNext;
        }

        @Override
        public T next() {

            try {
                // If hasNext has not been initialized, that means we are before
                // first row of the resultSet.
                // So we move the resultSet to the first row.
                if (hasNext == null) {
                    doNext();
                }

                // Fill the bean
                if (hasNext) {
                    T bean = null;
                    if (returnColumnName != null) {

                        // We have a specified output parameter to return,
                        // and ignore the rest.

                        JavaToJavaConverter explicitConverter = queryMetaData.getColumnConverter(returnColumnName);

                        Object returnColumnValue = resultSet.getObject(returnColumnName);
                        bean = bindResultToSimpleValue(returnedClass, returnColumnName, returnColumnValue, explicitConverter);

                    } else {

                    	Object discriminator = beanClassFactory == null ? null : resultSet.getObject(beanClassFactory.getDiscriminator());

                    	// If there is a bean factory, we create a bean using the class returned by it
                    	if (beanClassFactory != null) {

                    		@SuppressWarnings("unchecked")
							Class<? extends T> beanClass = (Class<? extends T>) beanClassFactory.getBeanClass(discriminator);

                    		bean = BeanUtils.newInstance(beanClass);
                    	} else {
                    		bean = BeanUtils.newInstance(returnedClass);
                    	}

                        for (int i = 1, size = resultSetMetadata.getColumnCount(); i <= size; i++) {

                            String columnLabel = resultSetMetadata.getColumnLabel(i);

                            if (paginationSupport == null
                                    || !paginationSupport.isColumnIgnored(SQLVariation.RANGED_QUERY, columnLabel)) {

                                if (!queryMetaData.isResultIgnored(columnLabel)
                                    && (beanClassFactory == null || !beanClassFactory.isResultIgnored(columnLabel, discriminator))) {

                                    JavaToJavaConverter explicitConverter = queryMetaData.getColumnConverter(columnLabel);
                                    if (beanClassFactory != null && explicitConverter == null) {
                                    	// If we have a beanClassFactory,
                                        // then the factory may return an explicit converter for a column for a bean.
                                    	explicitConverter = beanClassFactory.getResultConverter(columnLabel, discriminator);
                                    }

                                    Object resultValue = resultSet.getObject(columnLabel);
                                    String resultName = queryMetaData.getPropertyPathForResult(columnLabel);

                                    bindResultToBean(bean, resultName, resultValue, itermediatePropertyClasses, explicitConverter);
                                } else if (Config.isDebugMode()) {
                                    logger.info("Ignoring property binding for column: " + columnLabel);
                                }
                            }
                        }
                    }

                    // Move the resultSet, to next row, so next call to this
                    // method would return the next row.
                    doNext();

                    // And return the bean
                    return bean;

                } else {
                    throw new NoSuchElementException();
                }
            } catch (NoSuchElementException ex) {
                throw ex;
            } catch (Throwable t) {
                throw new SQLExecutionException(t);
            }
        }

        @Override
        public void remove() {
            // Removal is not supported
            throw new UnsupportedOperationException();
        }

        /**
         * Executes SELECT COUNT() on this {@code Query}'s sql statement in order to
         * obtain total rows returned by this query.
         *
         * @return total number of rows returned by this {@code Query}'s sql
         *         statement
         * @throws DataBindingException
         */
        private long getTotalRowsWithSelectCount() {

            PreparedStatement statement = null;
            ResultSet rs = null;

            try {

                NamedParametersHelper named = new NamedParametersHelper(paginationSupport.getSql(SQLVariation.COUNT_QUERY));

                if (Config.isDebugMode()) {
                    logger.info("Executing COUNT query:\n" + named.getSql());
                }

                statement = getConnection().prepareCall(named.getSql(), ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
                bindInputParameters(statement, named);

                // Binding additional parameters (for query-based range pagination)
                Collection<InputParameter> additionalParameters = paginationSupport.getAdditionalParameters(SQLVariation.COUNT_QUERY);
                if (paginationMode == PaginationMode.QUERY_BASED
                        && !additionalParameters.isEmpty()) {

                    for (InputParameter parameter : additionalParameters) {
                        bindInputParameter(statement, parameter, named);
                    }
                }

                long rowCount = 0;
                long tstart = System.currentTimeMillis();
                rs = statement.executeQuery();
                if (Config.isDebugMode()) {
                    logger.info("Executing COUNT query took: " + (System.currentTimeMillis() - tstart) + "ms.");
                }

                if (rs.next()) {
                    rowCount = rs.getLong(1);
                }

                return rowCount;
            } catch (Throwable t) {
                throw new SQLExecutionException(t);
            } finally {
                JDBCUtils.close(rs);
                JDBCUtils.close(statement);
            }
        }

    }

    /**
     * Determines if the query is paginated. The query is paginated if there is
     * a {@link RangeCallback} instance passed to the query, and defines
     * boundary other than {@code [1, infinity)}.
     *
     * @return true, if the query is paginated, false otherwise
     */
    private boolean isPaginated() {
        return rangeCallback != null && (rangeCallback.getStartRow() > 1 || rangeCallback.getStopRow() != -1);
    }

    /**
     * Utility method for binding input parameters to JDBC statement.
     *
     * @param stmt
     *            JDBC statement
     * @param named
     *            Named parameters helper object, for proper parameter binding
     * @throws DataBindingException
     *             if binding fails (for example due to lack of converter for
     *             unknown type)
     */
    private void bindInputParameters(final PreparedStatement stmt, final NamedParametersHelper named) throws DataBindingException {
        // Set parameters
        for (String parameterName : queryMetaData.getInputParameterNames()) {
            InputParameter parameter = queryMetaData.getInputParameter(parameterName);
            bindInputParameter(stmt, parameter, named);
        }
    }

    /**
     * Closes iterator.
     *
     * @param iterator
     *            iterator to close
     */
    private void closeIterator(final Iterator<?> iterator) {
        if (iterator != null) {
            iterator.close();
        }
    }

}
